package org.alis.rpc.registry;

import com.google.common.collect.Lists;
import lombok.extern.slf4j.Slf4j;
import org.alis.rpc.common.contant.ServiceConstant;
import org.alis.rpc.common.contant.ZookeeperConstant;
import org.alis.rpc.common.protocol.RpcProtocol;
import org.alis.rpc.common.protocol.RpcServiceInfo;
import org.alis.rpc.common.zookeeper.CuratorClient;
import org.apache.commons.lang3.StringUtils;
import org.apache.curator.framework.state.ConnectionState;

import java.util.List;
import java.util.Map;

/**
 * ServerRegistry desc:
 *
 * @author luochuan
 * @date 2020/9/21
 */
@Slf4j
public class ServerRegistry  {

    /**
     * 客户端配置
     */
    private CuratorClient curatorClient;

    /**
     * 路径集合
     */
    private List<String> pathList = Lists.newArrayList();

    public ServerRegistry(String registryAddress) {
        this.curatorClient = new CuratorClient(registryAddress, 5000);
    }


    /**
     * 注册服务
     * @param host 目标
     * @param port 端口
     * @param serviceMap 服务集合
     */
    public void registerService(String host, int port, Map<String, Object> serviceMap) {
        // Register service info
        List<RpcServiceInfo> serviceInfoList = Lists.newArrayList();
        serviceMap.forEach((k,v)->{
            String[] splitKey = StringUtils.split(k, ServiceConstant.SERVICE_CONCAT_TOKEN);
            if(splitKey.length>0){
                RpcServiceInfo rpcServiceInfo = new RpcServiceInfo();
                rpcServiceInfo.setServiceName(splitKey[0]);
                // 版本号放置正则版本之后的第三位
                int index = 2;
                if (splitKey.length == index) {
                    rpcServiceInfo.setVersion(splitKey[1]);
                } else {
                    rpcServiceInfo.setVersion("");
                }
                log.info("Register new service: {} ", k);
                serviceInfoList.add(rpcServiceInfo);
            }else{
                log.warn("can no get service name and the version");
            }
        });

        try{
            RpcProtocol rpcProtocol = new RpcProtocol();
            rpcProtocol.setHost(host);
            rpcProtocol.setPort(port);
            rpcProtocol.setServiceInfoList(serviceInfoList);
            String serviceData = rpcProtocol.toJson();
            byte[] bytes = serviceData.getBytes();
            String path = ZookeeperConstant.ZK_DATA_PATH + "-" + rpcProtocol.hashCode();
            this.curatorClient.createPathData(path, bytes);
            pathList.add(path);
            log.info("Register {} new service, host: {}, port: {}", serviceInfoList.size(), host, port);
        }catch (Exception e){
            log.error("Register service fail, exception: {}，throwable ->", e.getMessage(),e);
        }
        curatorClient.addConnectionStateListener(((curatorFramework, connectionState) -> {
            if (connectionState == ConnectionState.RECONNECTED) {
                log.info("Connection state: {}, register service after reconnected", connectionState);
                registerService(host, port, serviceMap);
            }
        }));
    }

    public void unregisterService() {
        log.info("Unregister all service");
        for (String path : pathList) {
            try {
                this.curatorClient.deletePath(path);
            } catch (Exception ex) {
                log.error("Delete service path error: {},throwable is ->" , ex.getMessage(),ex);
            }
        }
        this.curatorClient.close();
    }
}